package cn.iocoder.cloud.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import jakarta.websocket.SendResult;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;

/**
 * @author dengxiudong
 * @Date 2024/7/19 下午3:31
 */
@Slf4j
@Component
public class Producer
{

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;

    // 建议正常规模项目统一用一个TOPIC
    private static final String topic = "TEST_TOPIC";

    // 直接注入使用，用于发送消息到broker服务器
    @Resource
    private RocketMQTemplate rocketMQTemplate;


    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     * （msgBody也可以是对象，sendResult为返回的发送结果）
     */
    public SendResult sendMsg(String msgBody) {
        // SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build());
        // Message<String> message = MessageBuilder.withPayload(msgBody).setHeader("devId", "test123456").build();
        // 将消息发送到指定的Topic和Tag
        // rocketMQTemplate.send(topic, message);

        // log.info("【sendMsg】sendResult={}", JSON.toJSONString(sendResult));

        // return sendResult;

        String message = "test message";
        var result = rocketMQTemplate.syncSend("testTopic", message);
        log.info("【sendMsg】sendResult={}", JSON.toJSONString(result));
        return null;
    }

    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
     * （适合对响应时间敏感的业务场景）
     */
    // public void sendAsyncMsg(String msgBody) {
    //     rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
    //         @Override
    //         public void onSuccess(SendResult sendResult) {
    //             // 处理消息发送成功逻辑
    //         }
    //         @Override
    //         public void onException(Throwable throwable) {
    //             // 处理消息发送异常逻辑
    //         }
    //     });
    // }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public void sendDelayMsg(String msgBody, int delayLevel) {
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build(), messageTimeOut, delayLevel);
    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    public void sendOneWayMsg(String msgBody) {
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    public void sendTagMsg(String msgBody) {
        // return rocketMQTemplate.syncSend(topic + ":tag2", MessageBuilder.withPayload(msgBody).build());
        Message<String> message = MessageBuilder.withPayload(msgBody).setHeader("devId", "send123456").build();
        // 将消息发送到指定的Topic和Tag
        org.apache.rocketmq.client.producer.SendResult result = rocketMQTemplate.syncSendOrderly(topic, message, "send123456", 6000);
        log.info("【sendMsg】sendResult={}", JSON.toJSONString(result));
    }

    /***
     * 服务生产者，顺序消息
     * 把消息确保投递到同一条queue
     * 保证了消息的顺序性
     */
    public void sendFIFOMsg(List<String> users) {
        //顺序消息
        //选择器规则构建
        rocketMQTemplate.setMessageQueueSelector((list, message, o) -> {
            int id = Integer.valueOf((String) o);
            int hash = (id % list.size());
            return list.get(hash);
        });
        if (!CollectionUtils.isEmpty(users)) {
            for (String user : users) {
                MessageBuilder.withPayload(users.toString()).build();
                rocketMQTemplate.sendOneWayOrderly(topic+":sendFIFOMsg", user, String.valueOf(user));
            }
        }
    }

}

